package com.sunday.common.reactor.e5_Peeking_Into_Sequence;

import com.sunday.common.reactor.SubscribeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * A.3. 窥视序列
 */
@Slf4j
public class PeekingIntoSequence {

    /**
     * 排放：（通量|Mono#doOnNext)
     * 通过使用 doOnNext() 操作符，您可以在每次发出元素时执行一些操作，而不会改变元素本身。
     */
    @Test
    public void doOnNext() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .doOnNext(i -> log.info("[doOnNext] : {}", i));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 完成：Flux#doOnComplete、Mono#doOnSuccess（包括结果，如果有）
     * 通过使用 doOnComplete() 操作符，您可以在序列完成时执行一些操作。
     */
    @Test
    public void doOnComplete() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .doOnComplete(() -> log.info("[doOnComplete]"));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 错误终止：（Flux|Mono#doOnError)
     * 通过使用 doOnError() 操作符，您可以在序列以错误终止时执行一些操作。
     */
    @Test
    public void doOnError() {
        Flux<String> flux = Flux.just(1, 2, 0)
                .map(i -> "100 / " + i + " = " + (100 / i))
                .doOnError(error -> log.info("[doOnError] : {}", error.toString()));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 取消：（Flux|Mono#doOnCancel)
     * 通过使用 doOnCancel() 操作符，您可以在序列被取消时执行一些操作。
     */
    @Test
    public void doOnCancel() throws InterruptedException {
        Flux<Integer> flux = Flux.range(0, 10)
                .delayElements(Duration.ofSeconds(1))
                .doOnCancel(() -> log.info("[doOnCancel]"));
        Disposable disposable = flux.subscribe(i -> log.info("subscribe : {}", i),
                error -> log.error("error : {}", error.toString()),
                () -> log.info("Done =============="));

        log.info("[disposable.isDisposed()] {}", disposable.isDisposed());

        TimeUnit.SECONDS.sleep(5);
        disposable.dispose();
        log.info("[disposable.isDisposed()] {}", disposable.isDisposed());

        TimeUnit.SECONDS.sleep(10);

    }

    /**
     * 序列的“开始”：（Flux|Mono#doFirst)
     * 这与 Publisher#subscribe（Subscriber） 相关联
     * 通过使用 doFirst() 操作符，您可以在序列开始时执行一些操作。
     */
    @Test
    public void doFirst() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .doFirst(() -> log.info("[doFirst]"));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 订阅后 ： （Flux|Mono#doOnSubscribe)
     * Subscription之后的确认subscribe
     * 这与 Subscriber#onSubscribe（Subscription） 相关联
     * 通过使用 doOnSubscribe() 操作符，您可以在订阅时执行一些操作。
     */
    @Test
    public void doOnSubscribe() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .doOnSubscribe(subscription -> log.info("[doOnSubscribe] : {}", subscription.toString()));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 请求：（Flux|Mono#doOnRequest)
     * 通过使用 doOnRequest() 操作符，您可以在请求时执行一些操作。
     */
    @Test
    public void doOnRequest() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .doOnRequest(n -> log.info("[doOnRequest] : {}", n));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 完成或错误：（Flux|Mono#doOnTerminate)
     * 但在它被传播到下游之后：（Flux|Mono#doAfterTerminate)
     */
    @Test
    public void doOnTerminate() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                // doOnTerminate 在序列完成或错误时执行
                .doOnTerminate(() -> log.info("[doOnTerminate]"))
                // doAfterTerminate 在序列完成或错误时执行，但在它被传播到下游之后
                .doAfterTerminate(() -> log.info("[doAfterTerminate]"));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 任何类型的信号，表示为 Signal：（Flux|Mono#doOnEach)
     * 通过使用 doOnEach() 操作符，您可以在序列中的任何类型的信号上执行一些操作。
     */
    @Test
    public void doOnEach() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .doOnEach(signal -> log.info("[doOnEach] : {}", signal.toString()));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 任何终止条件（完成、错误、取消）：（Flux|Mono#doFinally)
     * 通过使用 doFinally() 操作符，您可以在序列终止时执行一些操作。
     */
    @Test
    public void doFinally() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .doFinally(signalType -> log.info("[doFinally] : {}", signalType.toString()));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 记录内部发生的情况：（Flux|Mono#log)
     * 通过使用 log() 操作符，您可以在序列中的任何类型的信号上执行一些操作。
     */
    @Test
    public void log() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                .log();
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 我想知道所有事件：
     * 每个都表示为 Signal 对象：
     * 而不是原来的 onNext 排放：（Flux|Mono#materialize)
     * 并返回 onNexts：（Flux|Mono#dematerialize)
     */
    @Test
    public void materialize() {
        Flux<Integer> flux = Flux.just(1, 2, 3)
                // materialize 将序列中的每个事件转换为信号对象
                .materialize()
                .doOnEach(signal -> log.info("[doOnEach] : {}", signal.toString()))
                .log()
                // dematerialize 将信号对象转换回序列中的每个事件
                .dematerialize();
        SubscribeUtils.subscribe(flux);
    }

}
